package com.offcn.bigdata.spark.p1.p3

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD

import scala.reflect.ClassTag

/*
* 二次排序：排序字段不唯一
*   当第一级的比较结果相同，再按照次一级的字段进行比较
* */
object _07SecondSortOps {
    def main(args: Array[String]): Unit = {
        val conf = new SparkConf()
            .setMaster("local[*]")
            .setAppName(s"${_07SecondSortOps.getClass.getSimpleName}")
        val sc = new SparkContext(conf)

        val list = List(
            Student(1, "林博", 18, 180),
            Student(2, "单松", 19, 150),
            Student(3, "张皓", 20, 120),
            Student(4, "王建", 20, 119),
            Student(106, "冯岩", 30, 10086)
        )
        val rdd = sc.parallelize(list)
        //按照学生的年龄进行排序,年龄相等按照身高排序
        val age2Stu:RDD[(Student, Option[Int])] = rdd.map(stu => (stu, None))

        age2Stu.sortByKey(ascending = true, numPartitions = 1).foreach(println)
        println("-----------------------------------------------")
        rdd.sortBy(stu => stu, ascending = true, numPartitions = 1)(
            new Ordering[Student](){
                override def compare(x: Student, y: Student): Int = {
                    var ret = y.height.compareTo(x.height)
                    if(ret == 0) {
                        ret = x.age.compareTo(y.age)
                    }
                    ret
                }
            },
            ClassTag.Object.asInstanceOf[ClassTag[Student]]
        ).foreach(println)
        sc.stop()
    }

    case class Student(id: Int, name: String, age: Int, height: Double) extends Ordered[Student] {
        override def compare(that: Student): Int = {
            var ret = this.age.compareTo(that.age)
            if(ret == 0) {
                ret = that.height.compareTo(this.height)
            }
            ret
        }
    }
}
